with network_massage as (
    select code
         , name
         , data
         , size(split(in_distribution_code_new, ',')) as size_num
    from jms_dim.dim_network_whole_massage
             lateral view explode(split(in_distribution_code_new, ',')) view1 as data
    where in_distribution_code_new is not null
      and in_distribution_code_new <> ''
      and name not like 'BEST%'
      and network_type = 6
),
     wide_unsign as (
         select count(waybill_no) as waybill_num
              , first_network_network_code
              , first_center_network_code
         from jms_dwd.dwd_wide_unsign_summary_waybill_dt
         where dt <= '{{ execution_date | cst_ds }}'
           and dt >= date_add('{{ execution_date | cst_ds }}', -6)
         and first_network_network_code is not null
         and first_center_network_code is not null
         group by first_network_network_code
                , first_center_network_code
     )
insert
overwrite
table
jms_dm.dm_tms_network_center_distri_dt
partition
(
dt
)
select code
     , name
     , center_code
     , center_name
     , size_num
     , waybill_num
     , '{{ execution_date | cst_ds }}' as date_time
     , '{{ execution_date | cst_ds }}' as dt
from (
         select t1.code
              , t3.name
              , t1.data                                                                                      as center_code
              , t4.name                                                                                      as center_name
              , t1.size_num
              , nvl(t1.waybill_num, 0)                                                                       as waybill_num
              , row_number() over (partition by t1.code order by nvl(t1.waybill_num, 0) desc ,t3.name desc ) as rn
         from (
                  select code
                       , data
                       , max(size_num)    as size_num
                       , sum(waybill_num) as waybill_num
                  from (
                           select code
                                , data
                                , size_num
                                , 0 as waybill_num
                           from network_massage
                           union all
                           select first_network_network_code as code
                                , first_center_network_code  as data
                                , 1                          as size_num
                                , waybill_num
                           from wide_unsign
                       ) a
                  group by code, data
              ) t1
         left join jms_dim.dim_network_whole_massage t3
                   on t1.code = t3.code
         left join jms_dim.dim_network_whole_massage t4
                   on t1.data = t4.code
     ) t
where rn = 1
and center_code is not null
    distribute by dt

